// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.utils.hbase;

import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.TSDB;
import net.opentsdb.utils.Assert;
import org.hbase.async.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public abstract class HbaseAbsUtils implements HbaseUtils{

    private static final Logger LOG = LoggerFactory.getLogger(HbaseAbsUtils.class);

    protected final HBaseClient client;
    protected final byte[] table;

    public static final Charset CHARSET = StandardCharsets.UTF_8;
    public static final byte[] EMPTY_ARRAY = new byte[0];

    public HbaseAbsUtils(HBaseClient client, byte[] table){
        this.client = client;
        this.table = table;
    }

    /**
     * 扫描表所有的数据
     *      TODO 应该加上scanner和rows限制否则数据量过大将会导致崩溃
     *
     * @param family
     * @return
     */
    public ArrayList<ArrayList<KeyValue>> scanTableData(final byte[] family){

        ArrayList<ArrayList<KeyValue>> results = new ArrayList<ArrayList<KeyValue>>();
        ArrayList<ArrayList<KeyValue>> rows;
        Scanner scanner = client.newScanner(table);

        try{
            scanner.setFamily(family);

            while ((rows=scanner.nextRows().joinUninterruptibly()) != null){
                results.addAll(rows);
            }
        }catch (Exception e){
            LOG.error("error scan table:",e);
        }finally {
            if(scanner != null)
                scanner.close();
            return results;
        }

    }


    @Override
    public void storeData(byte[] rowkey, byte[] families, byte[] qulitifer, byte[] value) {
        Assert.notNull(rowkey,"tsuid con't be null");

        final PutRequest tracking = new PutRequest(
                table,
                rowkey,
                families,
                qulitifer,
                value);

        client.put(tracking);
    }

    @Override
    public Deferred<Object> deleteData(byte[] rowkey, byte[] families, byte[] qulitifer) {
        Assert.notNull(rowkey,"key con't be null");
        // allow qulitifer be null
//        Assert.notNull(qulitifer,"qulitifer con't be null");

        final DeleteRequest delete = new DeleteRequest(
                table,
                rowkey,
                families,
                qulitifer);

        return client.delete(delete);
    }

    @Override
    public boolean existsData(byte[] rowkey, byte[] families, byte[] qulitifer) {
        Assert.notNull(rowkey,"rowkey con't be null");

        // 默认存在，不作操作
        boolean flag = true;

        final GetRequest get = new GetRequest(table, rowkey);
        get.family(families);

        if(qulitifer != null) get.qualifier(qulitifer);

        try{
            flag = client.get(get).addCallback(new ExistsCB()).joinUninterruptibly();
        }catch (Exception e){
            LOG.error(" get rowkey {} int tag table failed",rowkey,e);
        }

        return flag;
    }

    @Override
    public List<KeyValue> fetchData(byte[] rowkey, byte[] families, byte[] qulitifer) {
        Assert.notNull(rowkey,"rowkey con't be null");

        List<KeyValue> rows = new ArrayList<>();

        final GetRequest get = new GetRequest(table, rowkey);
        if(families != null) get.family(families);

        if(qulitifer != null) get.qualifier(qulitifer);

        try{
            rows = client.get(get).joinUninterruptibly();
        }catch (Exception e){
            LOG.error(" get rowkey {} int tag table failed",rowkey,e);
        }

        return rows;
    }

    @Override
    public boolean casStoreData(byte[] rowkey, byte[] families, byte[] qulitifer,
                                byte[] value, byte[] expected) {
        Assert.notNull(rowkey,"put request con't be null");
        if(expected == null) expected = EMPTY_ARRAY;
        boolean flag = true;

        final PutRequest put = new PutRequest(
                table,
                rowkey,
                families,
                qulitifer,
                value);

        try{
            flag = client.compareAndSet(put, expected).joinUninterruptibly();
        }catch (Exception e){
            LOG.error("compareAndSet put error:",e);
        }

        return flag;
    }

    final class ExistsCB implements Callback<Boolean, ArrayList<KeyValue>> {

        @Override
        public Boolean call(ArrayList<KeyValue> row){
            if (row == null || row.isEmpty() || row.get(0).value() == null) {
                return false;
            }
            return true;
        }

    }
}
